package com.xxpt.common.utils.amqp.config;

import com.xxpt.common.utils.amqp.util.CustomMessageConvert;
import com.xxpt.common.utils.amqp.util.RabbitMqUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.CustomExchange;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Import;

import java.util.HashMap;
import java.util.Map;

/**
 * rabbitmq配置类
  @author wlf
 * @mender icetea
 */
@Slf4j
@Import({MqConfig.class})
public class RabbitConfig {

    @Bean("directExchange")
    public DirectExchange directExchange() {
        log.info("声明direct交换器{}", RabbitMqUtil.EXCHANGE_DIRECT);
        return new DirectExchange(RabbitMqUtil.EXCHANGE_DIRECT, true, false);
    }

    /**
     * 延迟交换机
     * @return
     */
    @Bean("delayExchange")
    public CustomExchange delayExchange() {
        Map<String, Object> args = new HashMap<>();
        // 设置类型，可以为fanout、direct、topic
        args.put("x-delayed-type", "direct");
        // 第一个参数是延迟交换机名字，第二个是交换机类型，第三个设置持久化，第四个设置自动删除，第五个放参数
        return new CustomExchange(RabbitMqUtil.EXCHANGE_DELAYED_DIRECT,"x-delayed-message", true,false, args);
    }

    @Bean("topicExchange")
    public TopicExchange topicExchange() {
        log.info("声明topic交换器{}", RabbitMqUtil.EXCHANGE_TOPIC);
        return new TopicExchange(RabbitMqUtil.EXCHANGE_TOPIC, true, false);
    }

    @Bean("dlx")
    public DirectExchange dlx() {
        log.info("声明direct交换器{},用于路由死信队列", RabbitMqUtil.DEFAULT_EXCHANGE_DLX);
        return new DirectExchange(RabbitMqUtil.DEFAULT_EXCHANGE_DLX, true, false);
    }

    @Bean
    public MessageConverter messageConverter() {
        log.info("声明自定义rabbitMQ消息转换器");
        return new CustomMessageConvert();
    }

    /**
     * 注入自定义mq工具类
     * @return
     */
    @Bean
    public RabbitMqUtil getRabbitMqUtil(ConnectionFactory connectionFactory, MessageConverter messageConverter) {
        log.info("声明rabbitMQ自定义工具类");
        return new RabbitMqUtil(getRabbitTemplate(connectionFactory, messageConverter, true), getRabbitTemplate(connectionFactory, messageConverter, false));
    }

    public RabbitTemplate getRabbitTemplate(ConnectionFactory connectionFactory, MessageConverter messageConverter, boolean mandatory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        //connectionFactory.setPublisherConfirms(); CachingConnectionFactory
        rabbitTemplate.setConnectionFactory(connectionFactory);
        rabbitTemplate.setMessageConverter(messageConverter);   // 自定义消息转换器
        /*
         * 当mandatory标志位设置为true时
         * 如果exchange根据自身类型和消息routingKey无法找到一个合适的queue存储消息
         * 那么broker会调用basic.return方法将消息返还给生产者
         * 当mandatory设置为false时，出现上述情况broker会直接将消息丢弃
         */
        rabbitTemplate.setMandatory(mandatory); // 等同于 spring.rabbitmq.publisher-returns=true # 消息到达队列的回调
        // 设置配置回调函数
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                if(!ack){
                    log.info("ConfirmCallback:" + "相关数据：" + correlationData + ",确认情况：" + ack + ",原因：" + cause);
                    // 发送失败进行重试机制
                }
            }
        });
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                log.info("ReturnCallback:" + "消息：" + new String(message.getBody())  + ",回应码：" + replyCode  + ",回应信息：" + replyText  + ",交换机：" + exchange  + ",路由键：" + routingKey);
                //路由失败重试
            }
        });
        return rabbitTemplate;
    }
}